package hefei

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

/**
  * Created by Administrator on 2017/4/21 0021.
  */
object TheExample {
  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)

    // Load my user data and parse into tuples of user id and attribute list
    val users = (sc.textFile("users.txt")
      .map(line => line.split(",")).map( parts => (parts.head.toLong, parts.tail) ))
//    val list1 = users.take(3)
//    //判断list1类型
//    println(list1.asInstanceOf[AnyRef].getClass.getSimpleName)
//    for(x <- list1){
////      println(x.asInstanceOf[AnyRef].getClass.getSimpleName)
//      print(x._1+" : ")
//      for(y <- x._2){
//        print(y+"         ")
//      }
//      println()
//    }


    // Parse the edge data which is already in userId -> userId format
    val followerGraph = GraphLoader.edgeListFile(sc, "followers.txt")
    println("展示followerGraph的顶点:")
    followerGraph.vertices.collect.foreach(println(_))

    // Attach the user attributes
    val graph = followerGraph.outerJoinVertices(users) {
      case (uid, deg, Some(attrList)) => attrList
      // Some users may not have attributes so we set them as empty
      case (uid, deg, None) => Array.empty[String]
    }
    println("展示graph的顶点:")
    graph.vertices.collect.foreach(println(_))

    // Restrict the graph to users with usernames and names(只要带有username和name的顶点，其余的舍弃了)
    val subgraph = graph.subgraph(vpred = (vid, attr) => attr.size == 2)

    // Compute the PageRank
    val pagerankGraph = subgraph.pageRank(0.001)

    // Get the attributes of the top pagerank users
    //下图中case后面的括号中，有三个参数，第一个表示顶点ID，第二个表示subgraph的顶点的属性，第三个表示pagerankGraph的顶点属性。
    val userInfoWithPageRank = subgraph.outerJoinVertices(pagerankGraph.vertices) {
      case (uid, attrList, Some(pr)) => (pr, attrList.toList)
      case (uid, attrList, None) => (0.0, attrList.toList)
    }

    println(userInfoWithPageRank.vertices.top(5)(Ordering.by(_._2._1)).mkString("\n"))
    sc.stop()
  }
}
